package flink.clickhouse;

import com.alibaba.fastjson.JSON;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Random;
import java.util.UUID;

import static org.apache.flink.table.api.Expressions.$;

/**
 * 利用flink-connector-clickhouse写入到clickhouse
 */
public class clickhouseConnectorDemo {

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setString(RestOptions.BIND_PORT, "8081-8089");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        env.setParallelism(1);

        EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env, settings);


        DataStreamSource<Student> studentDataStreamSource = env.addSource(new SourceFunction<Student>() {
            private boolean flag = true;

            @Override
            public void run(SourceContext<Student> ctx) throws Exception {
                Random random = new Random();
                int id = 0;
                while (id < 1000) {
                    id++;
                    Thread.sleep(1000);
                    String name = UUID.randomUUID().toString();
                    int age = random.nextInt(100);
                    ctx.collect(new Student(id, name, age));
                }
            }

            @Override
            public void cancel() {
                flag = false;
            }
        });

        String sql = "Insert into student(id,name,age) values(?,?,?)";

        studentDataStreamSource.print();
        studentDataStreamSource.addSink(
                JdbcSink.sink(
                        sql,
                        new CkSinkBuilder(),
                        new JdbcExecutionOptions.Builder().withBatchSize(10).build(),
                        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                                .withDriverName("ru.yandex.clickhouse.ClickHouseDriver")
                                .withUrl("jdbc:clickhouse://192.168.62.38:8123/default")
                                .withUsername("liujian")
                                .withPassword("")
                                .build()
                )
        );

        env.execute("Flink DataStream to ClickHouse Example");


    }



    public static class Student {
        private Integer id;
        private String name;
        private Integer age;

        public Student() {
        }

        public Student(Integer id, String name, Integer age) {
            this.id = id;
            this.name = name;
            this.age = age;
        }

        public Integer getId() {
            return id;
        }

        public void setId(Integer id) {
            this.id = id;
        }

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        public Integer getAge() {
            return age;
        }

        public void setAge(Integer age) {
            this.age = age;
        }

        @Override
        public String toString() {
            return JSON.toJSONString(this);
        }
    }

    private static class CkSinkBuilder implements JdbcStatementBuilder<Student> {
        @Override
        public void accept(PreparedStatement ps, Student student) throws SQLException {
            ps.setInt(1,student.getId());
            ps.setString(2,student.getName());
            ps.setInt(3,student.getAge());

        }
    }
}
